package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object Demo22Acc {
  def main(args: Array[String]): Unit = {
    // 累加器
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo22Acc")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    // 通过List构建RDD
    val intRDD: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5), 2)

    var localVal: Int = 1

    /**
     * 在算子内部对算子外部的变量进行累加
     */
    intRDD
      .map(i => {
        /**
         * 算子内部的代码会以Task的形式发送到Executor执行
         * 如果在算子内部使用了算子外部的变量localVal
         * 则算子外部的变量localVal会以副本的形式发送到每个Executor
         *
         */
        println(s"原来的值：$localVal")
        localVal = localVal + 1
        println(s"+1的值：$localVal")
        i * i
      }).foreach(println)

    /**
     * 算子外部的代码在Driver端执行的
     * 所以在算子内部 对 算子外部的变量localVal 进行修改是不会生效的
     *
     */
    println(s"localVal的值为：$localVal")

    // 使用累加器
    // 在算子外部（Driver端）定义
    val longAcc: LongAccumulator = sc.longAccumulator("longAcc")
    intRDD
      .map(i => {
        // 在算子内部使用累加器
        println(s"longAcc原来的值：${longAcc.value}")
        longAcc.add(1)
        println(s"longAcc+1的值：${longAcc.value}")
        i * i
      }).foreach(println)

    // 在算子外部（Driver端）获取最终的结果
    println(s"longAcc的值为：${longAcc.value}")


  }

}
